package cn.itcast.documentapis;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * @Description TODO : BulkProcessor类提供了一个简单的接口，可根据请求的数量或大小自动刷新批量操作，或者在给定的时间段之后。
 * @Author: zj
 * @Date: 14:19 2019/3/8
 * @Version 1.0
 **/
public class BulkProcessorApi {


    @Test
    public void BulkProce() {
        TransportClient client = esClient.getClient();
        //可用于批量操作 和 测试api
        BulkProcessor build = BulkProcessor.builder(client, new BulkProcessor.Listener() {
            //在批量执行之前调用此方法
            public void beforeBulk(long l, BulkRequest bulkRequest) {

            }

            //批量执行后调用此方法
            public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {

            }

            //当批量失败并引发Throwable时，将调用此方法
            public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {

            }
        })
                //我们希望每10 000个请求执行批量处理
                .setBulkActions(10000)
                //我们想要每5mb冲洗一次
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                //无论请求数量多少，我们都希望每隔5秒刷新一次
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                //设置并发请求数。值为0表示只允许执行单个请求。值1表示允许执行1个并发请求，同时累积新的批量请求。
                .setConcurrentRequests(1)
                //设置自定义退避策略，该策略最初将等待100毫秒，以指数方式增加并重试最多三次。每当一个或多个批量项请求失败并且EsRejectedExecutionException指示可用于处理请求的计算资源太少时，就会尝试重试。
                // 要禁用退避，请传递BackoffPolicy.noBackoff（）。
                .setBackoffPolicy(
                        BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();
        //添加在 twitter 下 type doc 索引 为6 内容为 message
        build.add(new IndexRequest("twitter", "_doc", "6").source("message", "?sss"));
        //同事删除 twitter
        build.add(new DeleteRequest("twitter", "_doc", "2"));

        //当所有文档都加载到BulkProcessor时，可以使用awaitClose或close方法关闭它：
        try {
            build.awaitClose(10, TimeUnit.MINUTES);
            //or bulkProcessor.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        BulkRequest request = new BulkRequest();
        request.pipeline("globalId");

        request.add(new IndexRequest("twitter", "_doc", "1")
                .source(XContentType.JSON, "field", "bulk1")
                .setPipeline("perIndexId"));

        request.add(new IndexRequest("twitter", "_doc", "2")
                .source(XContentType.JSON, "field", "bulk2"));

    }
}
